package mws;


import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import mws.utils.JsonUtils;

import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

import static mws.ClientHolder.*;


@Setter
@Getter
public class ReduceServer {

    public static final String SERVER_NAME = ReduceServer.class.getName();

    private static final Integer INIT_ERROR = 8888;

    private static final Logger log = LoggerFactory.getLogger(ReduceServer.class);

    private static final ThreadLocal<Integer> channelType = new ThreadLocal<>();
    int port;
    // 创建主从Reactor线程组
// 主Reactor线程组(bossGroup)：负责监听服务器端口、接收客户端连接请求(TCP三次握手)
    EventLoopGroup bossGroup;

    // 从Reactor线程组(workerGroup)：负责处理已建立连接的IO事件
// 包括读取客户端发送的数据(OP_READ)、向客户端写入数据(OP_WRITE)等
    EventLoopGroup workerGroup;


    public ReduceServer() {
        this(8888);
    }

    public ReduceServer(int port) {

        this.port = port;
        try {
            initNettyServer();
        } catch (Exception e) {
            log.error("NettyServer init error", e);
            e.printStackTrace();
            System.exit(INIT_ERROR);
        }

    }

//    ServerSocketChannel 是 “总管”，只负责 “接客”（接受新连接）；
//    SocketChannel 是 “服务员”，每个服务员对应一个客人（客户端），专门负责和这个客人 “聊天”（处理数据）。
//    二者分工明确，是完全独立的 Channel 实例。

//    客户端 SocketChannel 关闭的核心场景：
//    客户端主动关闭（显式调用或进程退出）
//    服务器主动关闭（业务需求或心跳超时）
//    网络异常或中间设备干预
//            连接发生不可恢复的异常

    private void initNettyServer() throws Exception {

        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();

        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) {
                        // 添加处理器
                        ch.pipeline()

                                .addLast(new IdleStateHandler(10, 0, 0, TimeUnit.SECONDS),
                                        new HeartbeatHandler())

                                .addLast(// todo  这里的出站和入站编码器很重要的
                                new StringDecoder(Charset.forName("utf-8")),
                                new StringEncoder(Charset.forName("utf-8")),
                                new HelloHanlder(),
                                new RegisterAndSetThrealocalHolder(),
                                new ReadingAndSetChannel(),
                                new MessageMapperHandler()
                                // 第一个要实现的是 缓存所有客户端 然后 在断线的时候 还要删除客户端 （这个要看心跳 机制  那这里是对面主动发心跳还是我的一双向检测 我觉得是 ）
                        );
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true);


        // 绑定端口并启动服务器
        try {
            ChannelFuture future = bootstrap.bind(port).sync();
            future.channel().closeFuture().addListener(this::closeChannel);
        } catch (InterruptedException e) {
            throw new Exception(e);
        }
        System.out.println("服务器已启动，监听端口: " + port);

    }

    private void closeChannel(Future<? super Void> future) {
        System.out.println("端口 " + port + " 的服务器已关闭");
    }

    /**
     * 心跳超时处理器：检测客户端是否长时间未发送数据（包括心跳）
     */
    public class HeartbeatHandler extends ChannelInboundHandlerAdapter {

        // 当触发空闲事件时调用（由IdleStateHandler触发）
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;

                    // 检测到读空闲（30秒没收到客户端数据）
                    if (event.state() == IdleState.READER_IDLE) {
                        log.info("客户端 {} 已超过10秒未发送数据，关闭连接", ctx.channel().remoteAddress());
                        // 关闭通道（会触发 channelInactive 事件，可在此时清理客户端缓存）
                        ctx.close();
                    }

            } else {
                // 不是空闲事件，交给下一个处理器
                super.userEventTriggered(ctx, evt);
            }
        }
    }


    public class RegisterAndSetThrealocalHolder extends ChannelInboundHandlerAdapter {

        /**
         * 获取当前Channel的类型（所有入站事件处理前都会调用此方法）
         */
        private Integer getChannelType() {
            Integer type = channelType.get();
            if (type == null) {
                // 默认值或异常处理（根据业务需求）
                throw new IllegalStateException("Channel类型未初始化");
            }
            return type;
        }

        /**
         * 连接激活时初始化Channel类型（例如默认设为异步）
         * 第一次建立连接时，入站事件的触发顺序是：channelRegistered（Channel 注册到 EventLoop）→ channelActive（连接激活）。因此最先响应的是channelRegistered，而非channelActive。
         */

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            if (!ClientHolder.HavingChannel(ctx.channel())) {
                ClientHolder.addClient(ctx.channel(), DEFAULT);
            }
            // 初始化类型（实际场景可能需要根据客户端握手信息动态设置）
            channelType.set(ClientHolder.getClient(ctx.channel()));
            log.info("Channel激活，初始化类型为：" + (getChannelType() == DEFAULT ? "默认" : (getChannelType() == ASYNC ? "异步" : "同步")));
            super.channelRegistered(ctx);
        }


        /**
         * 处理读事件（所有入站事件的示例）
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            // 先获取ThreadLocal中的值
            channelType.set(ClientHolder.getClient(ctx.channel()));
            log.info("Channel激活，当前Channel类型：" + (getChannelType() == DEFAULT ? "默认" : (getChannelType() == ASYNC ? "异步" : "同步")));
            // 业务处理...
            super.channelRead(ctx, msg);
        }

        /**
         * 读完成事件
         */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            Integer type = getChannelType();
            log.info("读完成事件，当前Channel类型：" + (type == ASYNC ? "异步" : "同步"));
            super.channelReadComplete(ctx);
        }

        /**
         * 异常事件
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            Integer type = getChannelType();
            log.info("异常事件，当前Channel类型：" + (type == ASYNC ? "异步" : "同步"));
            super.exceptionCaught(ctx, cause);
        }

        /**
         * 连接关闭时清理ThreadLocal（关键：避免内存泄漏）
         */
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            log.info("Channel关闭，清理类型信息");
            channelType.remove(); // 必须清理，否则ThreadLocal可能持有过期数据
            super.channelInactive(ctx);
        }

        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            log.info("Channel取消注册，清理类型信息");
            ClientHolder.removeClient(ctx.channel());
            channelType.remove(); // 必须清理，否则ThreadLocal可能持有过期数据
            super.channelUnregistered(ctx);
        }
    }


    public class HelloHanlder extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("客户端已连接：" + ctx.channel().remoteAddress());
            ctx.writeAndFlush("欢迎连接到服务器！");
        }
    }

    public class ReadingAndSetChannel extends ChannelInboundHandlerAdapter {

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

//       todo     二、msg 如何实现结构化？

//            Netty 通过 编码（Encoder） 和 解码（Decoder） 处理器（属于 Pipeline 中的入站 / 出站处理器）实现 msg 的结构化，核心是解决「字节流 ↔ 业务对象」的转换问题。
//            1. 解码（入站方向：字节流 → 结构化对象）
//            当数据从网络到达服务器 / 客户端时，入站处理器中的解码器会将原始字节流（ByteBuf）转换为结构化对象：

            // 业务处理...
            //
            // Netty 的设计允许开发者根据需求选择 “自动传递”（super）或 “手动传递”（ctx.fireXXX），核心是保证事件在 Pipeline 中正确流转即可。


//            （2）手动传递（ctx.fireXXX）的典型场景
            //  todo “自动传递”（super）或 “手动传递”有什么区别呢

//            修改消息后传递：
//            如对消息进行加工后再传递给下一个处理器：
//            java
//                    运行
//            @Override
//            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//                String original = (String) msg;
//                String processed = original + "_processed"; // 加工消息
//                ctx.fireChannelRead(processed); // 传递修改后的消息（而非原始msg）
//                // 此处无需调用 super，否则会重复传递原始消息
//            }
//
//            传递多个消息：
//            如解码器将一个原始字节流拆分为多个结构化消息：
//            java
//                    运行
//            @Override
//            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//                ByteBuf buf = (ByteBuf) msg;
//                List<String> messages = splitIntoMultipleMessages(buf); // 拆分为多个消息
//                for (String msg : messages) {
//                    ctx.fireChannelRead(msg); // 逐个传递拆分后的消息
//                }
//                // 无需调用 super，否则会传递原始的 buf（已被拆分处理，无需再传递）
//            }
            String msgWrapper = (String) msg;
            MessageWrapper messageWrapper = new MessageWrapper(msgWrapper);
            if (!messageWrapper.isFormatRight()) {
                log.error("MapperClientHandler 收到未知消息类型：{}", messageWrapper.getType());
                ctx.writeAndFlush("wrapper:error:格式错误");
                return;
            }
            Channel channel = ctx.channel();
            ClientHolder.addClient(channel, messageWrapper.getType());
            ctx.fireChannelRead(messageWrapper);
        }

    }

    public class MessageMapperHandler extends ChannelInboundHandlerAdapter {

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (!(msg instanceof MessageWrapper )) {
                log.error("MapperClientHandler 收到未知消息类型 以及转化：{}", msg.getClass().getName());

            }
            MessageWrapper messageWrapper = (MessageWrapper) msg;

                if (messageWrapper.getType() == SYNC || messageWrapper.getType() == DEFAULT) {
                    String data = messageWrapper.getData();
                    storeMapper(data);
                    ctx.writeAndFlush("messageWrapper:success: 同步调用 或 DEFAULT同步" + messageWrapper.getData()).addListener(future -> {
                        if (future.isSuccess()) {
                            log.info("同步调用 或 DEFAULT同步 成功  回复成功");
                        }else {

                            log.error("同步调用 或 DEFAULT同步 失败  回复失败");
                            future.cause().printStackTrace();

                        }
                    });
                } else if (messageWrapper.getType() == ASYNC) {
                    ctx.writeAndFlush("messageWrapper:success:" + "异步调用成功").addListener(future -> {
                        if (future.isSuccess()) {
                            log.info("异步调用成功");
                            ClientHolder.removeClient(ctx.channel());
                            ctx.channel().close();
                        } else {
                            // 1. 第一步：详细日志记录（关键！用于排查）
                            Throwable cause = future.cause();
                            log.error(
                                    "消息发送失败！通道：{}，消息：{}，原因：{}",
                                    ctx.channel().id(),  // 通道唯一标识
                                    messageWrapper,  // 失败的消息内容
                                    cause.getMessage()    // 失败原因
                            );
                            cause.printStackTrace();  // 打印完整堆栈，定位代码问题

                            // 2. 第二步：强制清理资源（避免内存泄漏）
                            // 无论失败原因，都要移除客户端引用+关闭通道（防止无效连接占用资源）
                            ClientHolder.removeClient(ctx.channel());
                            if (ctx.channel().isOpen()) {  // 避免重复关闭
                                ctx.channel().close();
                            }

                            // 3. 第三步：根据失败原因，差异化补偿（核心）
                            handleSendFailure(ctx, "wrapper:success:异步调用成功", cause);
                        }
                    });
                } else {
                    log.error("MapperClientHandler 收到messageWrapper未知消息类型：{}", messageWrapper.getType());
                }

        }

        public void handleSendFailure(ChannelHandlerContext ctx, String s, Throwable cause) {
            // 异步这个消息就算发送失败也不用管
        }

        public void storeMapper( String data) {
            // 同步消息存储到缓存中
            HashMap map = JsonUtils.parseJson(data, HashMap.class);
            map.keySet().forEach(key -> {
                log.info("storeSyncMessage key:{} value:{}", key, map.get(key));
                MapperHolder.putMapper((String) key,Integer.valueOf((String) map.get(key)));
            });
        }
    }

}



